package com.hbase.zhou;

import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.junit.Test;

import java.io.IOException;
import java.util.Arrays;


/**
 * @Description:
 * @Author: ZhOu
 * @Date: 2018/3/7
 */
@Slf4j
public class HBaseTest {
    private static HBaseAdmin hBaseAdmin;
    private static Connection connection;

    static {
        Configuration configuration = HBaseConfiguration.create();
        configuration.set("hbase.zookeeper.property.clientPort", "2181");
        configuration.set("hbase.zookeeper.quorum", "node1,node2,node3");
        try {
            connection = ConnectionFactory.createConnection(configuration);
            hBaseAdmin = (HBaseAdmin) connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    private void close() {
        try {
            if (hBaseAdmin != null) {
                hBaseAdmin.close();
            }
            if (connection != null) {
                connection.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列出所有表
     */
    @Test
    public void listTable() throws IOException {
        TableName[] tb = hBaseAdmin.listTableNames();
        Arrays.asList(tb).forEach(v -> System.out.println(v));
        close();
    }

    /**
     * 新增表
     */
    @Test
    public void createTable() throws IOException {
        String tableName = "test_tb";
        if (hBaseAdmin.tableExists(tableName)) {
            hBaseAdmin.disableTable(tableName);
            hBaseAdmin.deleteTable(tableName);
        }

        HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
        htd.addFamily(new HColumnDescriptor("cf1"));
        HColumnDescriptor hcd = new HColumnDescriptor("cf2");
        //单元格数据的过期时间
        hcd.setTimeToLive(3);
        hcd.setMaxVersions(2);
        hcd.setMinVersions(1);
        htd.addFamily(hcd);

        hBaseAdmin.createTable(htd);
        System.out.println("test_tb表创建成功");
        close();

    }

    /**
     * 插入数据
     */
    @Test
    public void inputData() throws IOException {
        Table table = connection.getTable(TableName.valueOf("test_tb"));
        String rowKey = "t001";
        Put put = new Put(rowKey.getBytes());
        put.addColumn("cf1".getBytes(), "name".getBytes(), "Tony".getBytes());
        table.put(put);
        System.out.println("数据插入成功");
        table.close();
        close();
    }

    /**
     * 查询所有数据
     */
    @Test
    public void scanData() throws IOException {
        Table table = connection.getTable(TableName.valueOf("test_tb"));
        Scan scan = new Scan();
        scan.setId("t001");
        ResultScanner scanner = table.getScanner(scan);
        scanner.forEach(result -> {
            byte[] row = result.getRow();
            System.out.println("row=" + new String(row));
            result.listCells().forEach(cell ->
                    System.out.println("\t cf=" + new String(CellUtil.cloneFamily(cell))
                            + "\t column=" + new String(CellUtil.cloneQualifier(cell))
                            + "\t row=" + new String(CellUtil.cloneRow(cell))
                            + "\t value=" + new String(CellUtil.cloneValue(cell))
                            + "\t timestamp=" + cell.getTimestamp())
            );
        });
        scanner.close();
        table.close();
        close();
    }

    /**
     * 获取指定数据
     */
    @Test
    public void getOne() throws IOException {
        Table table = connection.getTable(TableName.valueOf("tb1"));
        Get get = new Get("t001".getBytes());
        Result result = table.get(get);
        byte[] row = result.getRow();
        System.out.println("row=" + new String(row));

        result.listCells().forEach(cell ->
                System.out.println("\t cf=" + new String(CellUtil.cloneFamily(cell))
                        + "\t column=" + new String(CellUtil.cloneQualifier(cell))
                        + "\t row=" + new String(CellUtil.cloneRow(cell))
                        + "\t value=" + new String(CellUtil.cloneValue(cell)))
        );

        table.close();
        close();
    }

    /**
     * 删除指定数据
     */
    @Test
    public void deleteOne() throws IOException {
        Table table = connection.getTable(TableName.valueOf("test_tb"));
        Delete delete = new Delete("t001".getBytes());
        //删除最新版数据
//        delete.addColumn("cf2".getBytes(),"name".getBytes());

        //删除所有版本的数据
        delete.addColumns("cf2".getBytes(),"name".getBytes());

        table.delete(delete);
        System.out.println("数据删除成功");
        table.close();
        close();
    }
}
